package com.chukun.multi.pipeline.reusable;

import java.util.LinkedList;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

public class SimplePipeline<T,OUT> extends AbstractPipe<T,OUT> implements Pipeline<T,OUT>  {

    private final Queue<Pipe<T,OUT>> pipes = new LinkedList<>();
    private final ExecutorService helperExecutor;

    public SimplePipeline() {
        this(Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r,"simplePipeline-helper");
                t.setDaemon(true);
                return t;
            }
        }));
    }
    public SimplePipeline(ExecutorService executorService) {
        super();
        this.helperExecutor = executorService;
    }

    @Override
    public void shutdown(long timeout, TimeUnit unit) {
        Pipe<T,OUT> pipe;
        while (!Objects.isNull(pipe=pipes.poll())) {
            pipe.shutdown(timeout,unit);
        }
        helperExecutor.shutdown();
    }

    @Override
    protected OUT doProcess(T input) throws InterruptedException {
        return null;
    }

    @Override
    public void addPipe(Pipe<T, OUT> pipe) {
        // Pipe间的关联关系在init中建立
       pipes.add(pipe);
    }

    @Override
    public void process(T input) throws InterruptedException {
        Pipe<T, OUT> firstPipe = pipes.peek();
        firstPipe.process(input);
    }

    @Override
    public void init(PipeContext pipeCtx) {
        LinkedList<Pipe<T,OUT>> pipeLinkedList = (LinkedList<Pipe<T, OUT>>) pipes;
        Pipe<T,OUT> prePipe = this;
        for (Pipe<T, OUT> pipe : pipeLinkedList) {
            prePipe.setNextPipe(pipe);
            prePipe = pipe;
        }
        Runnable task = ()->{
            for (Pipe<T, OUT> pipe : pipes) {
                pipe.init(pipeCtx);
            }
        };
        helperExecutor.submit(task);
    }

    public PipeContext newDefaultPipeContext() {
        return new PipeContext() {
            @Override
            public void handleError(PipeException e) {
                helperExecutor.submit(()->{
                    e.printStackTrace();
                });
            }
        };
    }
}
